package cn.itcast.kafka;

import cn.itcast.avro.AvroSerializer;
import org.apache.avro.specific.SpecificRecordBase;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

//表示传入对象时avro类型的对象
public class KafkaPro<T extends SpecificRecordBase> {

    KafkaProducer kafkaProducer = new KafkaProducer(getProp());

    /**
     * 1.创建类，泛型参数继承avro基类
     * 2.设置生产者参数
     * 3.自定avro序列化
     * 4.添加发送数据方法
     */
    public Properties getProp(){
        /**
         * bootstrap.servers ：broker地址
         * acks ：0,1和-1
         * retries：重试次数
         * batch.size：批量发送大小默认16384 （16kB）
         * linger.ms： 定时发送1ms
         * buffer.memory: 缓存大小33554432
         * key.serializer：key序列化
         * value.serializer： value序列化
         */
        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers","node01:9092");
        prop.setProperty("acks","0");//客户端只管发送，服务端是否收到不管
        prop.setProperty("retries","0");
        prop.setProperty("batch.size","16384");
        prop.setProperty("linger.ms","1");
        prop.setProperty("buffer.memory","33554432");
        prop.setProperty("key.serializer", StringSerializer.class.getName());
        prop.setProperty("value.serializer", AvroSerializer.class.getName());//自定义序列化
        return prop;
    }
    //发送数据
    public void sendData(String topic,T data){
        kafkaProducer.send(new ProducerRecord(topic,data));
    }

}
